package com.chinasoft.bigdatadisplay.utils;

import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.web.multipart.MultipartFile;

import java.io.*;
import java.util.ArrayList;
import java.util.List;

/**
 * HDFS操作工具类
 * 封装了常用的HDFS文件操作，如上传、下载、读取、删除等
 */
@Component
public class HdfsUtils {

    @Autowired
    private FileSystem fileSystem;

    @Value("${hadoop.core.fs.defaultFS}")
    private String hdfsUri;

/*    @Value("${hdfs.upload.base-path}")
    private String baseUploadPath;*/

    // 关键修改：将hdfs改为hadoop，与yml中的hadoop.upload.base-path匹配
    @Value("${hadoop.upload.base-path}")
    private String baseUploadPath;

    /**
     * 从HDFS读取文件内容并返回字符串列表
     * @param hdfsPath HDFS文件路径（如：/user/test.txt 或 hdfs://hn41:8020/user/test.txt）
     * @return 文件内容行列表
     */
    public List<String> readHdfsFile(String hdfsPath) throws IOException {
        List<String> contentList = new ArrayList<>();
        Path path = new Path(hdfsPath);
        
        // 检查文件是否存在
        if (!fileSystem.exists(path) || fileSystem.isDirectory(path)) {
            throw new FileNotFoundException("HDFS文件不存在或为目录: " + hdfsPath);
        }
        
        try (FSDataInputStream inputStream = fileSystem.open(path);
             BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream))) {
            
            String line;
            while ((line = reader.readLine()) != null) {
                contentList.add(line);
            }
        }
        
        return contentList;
    }

    /**
     * 上传本地文件到HDFS
     * @param localFilePath 本地文件路径
     * @param hdfsDir HDFS目标目录
     * @return 上传后的HDFS文件路径
     */
    public String uploadLocalFile(String localFilePath, String hdfsDir) throws IOException {
        Path localPath = new Path(localFilePath);
        Path hdfsDirectory = new Path(hdfsDir);
        
        // 检查本地文件是否存在
        File localFile = new File(localFilePath);
        if (!localFile.exists() || !localFile.isFile()) {
            throw new FileNotFoundException("本地文件不存在或为目录: " + localFilePath);
        }
        
        // 创建HDFS目录（如果不存在）
        if (!fileSystem.exists(hdfsDirectory)) {
            fileSystem.mkdirs(hdfsDirectory);
        }
        
        // 上传文件
        Path hdfsFilePath = new Path(hdfsDir + "/" + localFile.getName());
        fileSystem.copyFromLocalFile(localPath, hdfsFilePath);
        
        return hdfsFilePath.toString();
    }

    /**
     * 上传MultipartFile到HDFS（用于Web文件上传）
     * @param file 前端上传的文件
     * @return 上传后的HDFS文件路径
     */
    public String uploadMultipartFile(MultipartFile file) throws IOException {
        return uploadMultipartFile(file, baseUploadPath);
    }

    /**
     * 上传MultipartFile到HDFS指定目录
     * @param file 前端上传的文件
     * @param hdfsDir HDFS目标目录
     * @return 上传后的HDFS文件路径
     */
    public String uploadMultipartFile(MultipartFile file, String hdfsDir) throws IOException {
        if (file.isEmpty()) {
            throw new IOException("上传的文件为空");
        }
        
        Path hdfsDirectory = new Path(hdfsDir);
        // 创建HDFS目录（如果不存在）
        if (!fileSystem.exists(hdfsDirectory)) {
            fileSystem.mkdirs(hdfsDirectory);
        }
        
        // 上传文件流
        String fileName = file.getOriginalFilename();
        Path hdfsFilePath = new Path(hdfsDir + "/" + fileName);
        
        try (InputStream inputStream = file.getInputStream();
             FSDataOutputStream outputStream = fileSystem.create(hdfsFilePath)) {
            
            IOUtils.copyBytes(inputStream, outputStream, 4096, true);
        }
        
        return hdfsFilePath.toString();
    }

    /**
     * 从HDFS下载文件到本地
     * @param hdfsPath HDFS文件路径
     * @param localDir 本地目标目录
     * @return 本地文件路径
     */
    public String downloadFile(String hdfsPath, String localDir) throws IOException {
        Path hdfsFilePath = new Path(hdfsPath);
        File localDirectory = new File(localDir);
        
        // 检查HDFS文件是否存在
        if (!fileSystem.exists(hdfsFilePath) || fileSystem.isDirectory(hdfsFilePath)) {
            throw new FileNotFoundException("HDFS文件不存在或为目录: " + hdfsPath);
        }
        
        // 创建本地目录（如果不存在）
        if (!localDirectory.exists()) {
            localDirectory.mkdirs();
        }
        
        // 下载文件
        String fileName = hdfsFilePath.getName();
        Path localFilePath = new Path(localDir + File.separator + fileName);
        fileSystem.copyToLocalFile(hdfsFilePath, localFilePath);
        
        return localFilePath.toString();
    }

    /**
     * 删除HDFS上的文件或目录
     * @param hdfsPath 要删除的HDFS路径
     * @param recursive 如果是目录，是否递归删除
     * @return 是否删除成功
     */
    public boolean deleteHdfsPath(String hdfsPath, boolean recursive) throws IOException {
        Path path = new Path(hdfsPath);
        if (!fileSystem.exists(path)) {
            return false;
        }
        return fileSystem.delete(path, recursive);
    }

    /**
     * 列出HDFS目录下的所有文件
     * @param hdfsDir HDFS目录路径
     * @return 文件信息列表
     */
    public List<HdfsFileInfo> listFiles(String hdfsDir) throws IOException {
        List<HdfsFileInfo> fileInfoList = new ArrayList<>();
        Path dirPath = new Path(hdfsDir);
        
        if (!fileSystem.exists(dirPath) || !fileSystem.isDirectory(dirPath)) {
            return fileInfoList;
        }
        
        RemoteIterator<LocatedFileStatus> fileStatuses = fileSystem.listFiles(dirPath, false);
        while (fileStatuses.hasNext()) {
            LocatedFileStatus status = fileStatuses.next();
            HdfsFileInfo fileInfo = new HdfsFileInfo();
            fileInfo.setFileName(status.getPath().getName());
            fileInfo.setFilePath(status.getPath().toString());
            fileInfo.setFileSize(status.getLen());
            fileInfo.setModificationTime(status.getModificationTime());
            fileInfo.setReplication(status.getReplication());
            fileInfoList.add(fileInfo);
        }
        
        return fileInfoList;
    }

    /**
     * HDFS文件信息封装类
     */
    public static class HdfsFileInfo {
        private String fileName;
        private String filePath;
        private long fileSize;
        private long modificationTime;
        private short replication;

        // getter和setter方法
        public String getFileName() { return fileName; }
        public void setFileName(String fileName) { this.fileName = fileName; }
        public String getFilePath() { return filePath; }
        public void setFilePath(String filePath) { this.filePath = filePath; }
        public long getFileSize() { return fileSize; }
        public void setFileSize(long fileSize) { this.fileSize = fileSize; }
        public long getModificationTime() { return modificationTime; }
        public void setModificationTime(long modificationTime) { this.modificationTime = modificationTime; }
        public short getReplication() { return replication; }
        public void setReplication(short replication) { this.replication = replication; }
    }
}
